package com.bs.rocketmq.listen;

import com.alibaba.fastjson.JSON;
import com.bs.rocketmq.po.MessageDeduplication;
import com.bs.rocketmq.service.MessageDeduplicationService;
import com.bs.rocketmq.service.MyService;
import com.bs.rocketmq.vo.UserVo;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author bingshao
 * @date 2023/6/2
 **/
@Log4j2
@Component
@RocketMQMessageListener(
        topic = "hasKey-topic",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING,
        consumerGroup = "bs-hasKey-consumer")
public class ReliabilityConsumerListener implements RocketMQListener<MessageExt> {

    @Resource
    private MessageDeduplicationService messageDeduplicationService;

    @Resource
    private MyService myService;

    @Override
    public void onMessage(MessageExt messageExt) {
        log.info("同步接收到消息=====>{}", JSON.parseObject(new String(messageExt.getBody()), UserVo.class).toString());
        Long keys = Long.valueOf(messageExt.getKeys());
        MessageDeduplication messageDeduplication = new MessageDeduplication();
        messageDeduplication.setId(keys);
        try {
            messageDeduplicationService.save(messageDeduplication);
        } catch (Exception e) {
            log.info("消息已经消费过了===={}", JSON.parseObject(new String(messageExt.getBody()), UserVo.class).toString());
        }

        // 能插入数据说明未重复，直接处理业务即可
        try {
            myService.doSomeThing();
        } catch (Exception e) {
            // 消费失败删除去重表的记录
            messageDeduplicationService.removeById(keys);

            // 获取重试次数
            int reconsumeTimes = messageExt.getReconsumeTimes();
            // 设置重试两次
            if (reconsumeTimes > 2) {
                // TODO 将消息持久化到数据库，多次消费失败基本是业务代码有问题，需要人工干预

                return;
            }
            // 触发 RocketMQ 重试
            throw new RuntimeException("业务异常，触发消息重试", e);
        }

    }
}
